package com.study.mr.outputformat;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

/**
 * 自定义logRecordWriter，需要继承hadoop的RecordWriter
 */
public class LogRecordWriter extends RecordWriter<Text,NullWritable>{
    //定义输出路径
    private String atguiguPath="D:\\hadoopbook\\demodate\\log_out\\atguigu.txt";
    private String otherPath="D:\\hadoopbook\\demodate\\log_out\\other.txt";
    private FileSystem fs;
    private FSDataOutputStream atguiguOut;
    private FSDataOutputStream otherOut;

    /**
     * 初始化工作
     * @param job
     */
    public LogRecordWriter(TaskAttemptContext job) throws IOException {
        //获取hadoop文件系统对象
        fs=FileSystem.get(job.getConfiguration());
        //获取输出流对象atguiguOut
        atguiguOut = fs.create(new Path(atguiguPath));
        //获取输出流对象otherOut
        otherOut = fs.create(new Path(otherPath));
    }

    /**
     *实现数据写出逻辑
     * @param key
     * @param value
     * @throws IOException
     * @throws InterruptedException
     */
    public void write(Text key, NullWritable value) throws IOException, InterruptedException {
        //获取当前输入数据
        String lodData = key.toString();
        if (lodData.contains("atguigu")){
            atguiguOut.writeBytes(lodData+"\n");
        }else {
            otherOut.writeBytes(lodData+"\n");
        }

    }

    /**
     *关闭资源
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        IOUtils.closeStream(atguiguOut);
        IOUtils.closeStream(otherOut);
    }
}
